/*
 * Copyright 2012-2016 the original author or authors. Licensed under the Apache
 * License, Version 2.0 (the "License"); you may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 * http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law
 * or agreed to in writing, software distributed under the License is
 * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the specific language
 * governing permissions and limitations under the License.
 */

package com.neusoft.bizcore.mom.config;

import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.config.ContainerProperties;

import com.neusoft.bizcore.mom.config.KafkaProperties.Listener;

/**
 * Configure {@link ConcurrentKafkaListenerContainerFactory} with sensible
 * defaults.
 *
 * @author Gary Russell
 * @since 1.5.0
 */
public class ConcurrentKafkaListenerContainerFactoryConfigurer {

    private KafkaProperties properties;

    /**
     * Set the {@link KafkaProperties} to use.
     *
     * @param properties the properties
     */
    void setKafkaProperties(final KafkaProperties properties) {
        this.properties = properties;
    }

    /**
     * Configure the specified Kafka listener container factory. The factory can
     * be further tuned and default settings can be overridden.
     *
     * @param listenerContainerFactory the
     *        {@link ConcurrentKafkaListenerContainerFactory} instance to
     *        configure
     * @param consumerFactory the {@link ConsumerFactory} to use
     */
    public void configure(
            final ConcurrentKafkaListenerContainerFactory<Object, Object> listenerContainerFactory,
            final ConsumerFactory<Object, Object> consumerFactory) {
        listenerContainerFactory.setConsumerFactory(consumerFactory);
        final Listener container = this.properties.getListener();
        final ContainerProperties containerProperties = listenerContainerFactory
                .getContainerProperties();
        if (container.getAckMode() != null) {
            containerProperties.setAckMode(container.getAckMode());
        }
        if (container.getAckCount() != null) {
            containerProperties.setAckCount(container.getAckCount());
        }
        if (container.getAckTime() != null) {
            containerProperties.setAckTime(container.getAckTime());
        }
        if (container.getPollTimeout() != null) {
            containerProperties.setPollTimeout(container.getPollTimeout());
        }
        if (container.getConcurrency() != null) {
            listenerContainerFactory.setConcurrency(container.getConcurrency());
        }
    }

}
